package com.demo.rocketmq.producer;

import com.demo.rocketmq.config.ProducerConfig;
import com.demo.rocketmq.config.RocketMqConfig;
import com.demo.rocketmq.listener.DemoTransactionConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

/**
 * https://rocketmq.apache.org/zh/docs/4.x/producer/09message5/
 */
@Configuration
@Slf4j
public class DemoTransactionProducer {

    @Autowired
    private RocketMqConfig rocketMqConfig;
    @Autowired
    private ProducerConfig rocketMqProducerConfig;
    @Autowired
    private DemoTransactionConsumerListener demoTransactionConsumerListener;


    /**
     * 定义一个线程池
     * corePoolSize 线程池中核心线程数量
     * maximumPoolSize 线程池中最多线程数
     * keepAliveTime 这是一个时间。当线程池中线程数量大于核心线程数量时多余空闲线程的存活时长
     * unit 时间单位
     * workQueue 临时存放任务的队列，其参数就是队列的长度
     * threadFactory 线程工厂
     */
    private ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        }

    });

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public TransactionMQProducer buildGoodsTransactionProducer() {

        TransactionMQProducer producer;
        if (rocketMqConfig.getAclActive()) {
            // 生产环境使用阿里云mq
            producer = new TransactionMQProducer(null, rocketMqConfig.getDefaultProducerGroup(), rocketMqConfig.getAclRPCHook(), true, null);
            // 设置使用接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项。
            producer.setAccessChannel(AccessChannel.CLOUD);
        } else {
            producer = new TransactionMQProducer(rocketMqConfig.getDefaultProducerGroup());
        }
        producer.setNamesrvAddr(rocketMqConfig.getNamesrvAddr());
        producer.setMaxMessageSize(rocketMqProducerConfig.getMaxMessageSize());
        producer.setSendMsgTimeout(rocketMqProducerConfig.getSendMsgTimeout());
        producer.setRetryTimesWhenSendFailed(rocketMqProducerConfig.getRetryTimesWhenSendFailed());
        // 只能注册一个，最后的生效，所以需要多个生产者实例
        producer.setTransactionListener(demoTransactionConsumerListener);
        // 为生产者指定一个线程池
        producer.setExecutorService(executorService);
        log.info("《====================【rocketmq事务生产者启动成功】====================》");
        return producer;
    }

}